package org.example.rabbitmq.client.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeoutException;

public class RPCServer {
    CountDownLatch latch = new CountDownLatch(1);

    public void shutdown() {
        latch.countDown();
    }

    public void start() {
        Thread t = new Thread(() -> {
            Connection connection = null;
            try {
                //尝试获取一个连接
                connection = newConnection();
                //尝试创建一个channel
                final Channel channel = connection.createChannel();
                declareQueue(channel);
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                                .Builder()
                                .correlationId(properties.getCorrelationId())
                                .build();
                        String message = new String(body, "UTF-8");
                        String response = "handle message on server: " + message;
                        channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
                        channel.basicAck(envelope.getDeliveryTag(), false);
                        // ? RabbitMq consumer worker thread notifies the RPC server owner thread
                    }
                };
                channel.basicConsume(Names.RPC_QUEUE_NAME, false, consumer);
                System.out.println("[x] Awaiting RPC requests");
                latch.await();
            } catch (InterruptedException | IOException | TimeoutException e) {
                e.printStackTrace();
            } finally {
                closeConnection(connection);
            }
        });
        t.start();
    }

    private Connection newConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest@local");
        return factory.newConnection();
    }

    private void declareQueue(Channel channel) throws IOException {
        channel.queueDeclare(Names.RPC_QUEUE_NAME, false, false, false, null);
    }

    private void closeConnection(Connection connection) {
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}